package com.xjd.transport.impl;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.xjd.transport.Channel;
import com.xjd.transport.TransportClient;
import com.xjd.transport.TransportException;
import com.xjd.transport.TransportStates;
import com.xjd.transport.impl.ChannelImpl.TransportAdapter;

public class TransportClientImpl implements TransportClient, TransportStates {
	private static Logger log = LoggerFactory.getLogger(TransportClientImpl.class);

	public static int DEFAULT_MAX_OPENED_CHANNEL = 50;

	protected Object stateLock = new Object();
	protected int state = STATE_INITIAL;

	protected ExecutorService workerExecutorService;

	protected int maxOpenedChannel = DEFAULT_MAX_OPENED_CHANNEL;
	List<ChannelImpl> openedChannels = new LinkedList<ChannelImpl>();

	public TransportClientImpl() {
	}

	public TransportClientImpl(int maxOpenedChannel) {
		this.maxOpenedChannel = maxOpenedChannel;
	}

	public TransportClientImpl(int maxOpenedChannel, ExecutorService workerExecutorService) {
		this.maxOpenedChannel = maxOpenedChannel;
		this.workerExecutorService = workerExecutorService;
	}

	protected void checkStarted() {
		synchronized (stateLock) {
			if (state != STATE_INITIAL) {
				throw new IllegalStateException("TransportClient already started or stopped.");
			}
		}
	}

	@Override
	public void setWorkerExecutorService(ExecutorService workerExecutorService) {
		checkStarted();
		this.workerExecutorService = workerExecutorService;
	}

	@Override
	public ExecutorService getWorkerExecutorService() {
		return workerExecutorService;
	}

	@Override
	public void setMaxOpenedChannel(int maxOpened) {
		checkStarted();
		this.maxOpenedChannel = maxOpened;
	}

	@Override
	public int getMaxOpenedChannel() {
		return this.maxOpenedChannel;
	}
	
	@Override
	public Channel openChannel(String host, int port) throws TransportException {
		return openChannel(new InetSocketAddress(host, port));
	}

	@Override
	public Channel openChannel(InetSocketAddress destAddress) throws TransportException {
		synchronized (stateLock) {
			if (state == STATE_STOPED) {
				throw new IllegalStateException("TransportClient already stopped.");
			}
			if (state == STATE_INITIAL) {
				state = STATE_STARTED;
			}
			SocketChannel socketChannel;
			try {
				socketChannel = SocketChannel.open(destAddress);
			} catch (IOException e) {
				throw new TransportException(e);
			}
			ChannelImpl channel = new ChannelImpl(socketChannel, this.workerExecutorService, new ClientAdatper());
			synchronized (openedChannels) {
				openedChannels.add(channel);
			}
			return channel;
		}

	}

	@Override
	public Channel[] getOpenedChannels() {
		synchronized (openedChannels) {
			return openedChannels.toArray(new ChannelImpl[openedChannels.size()]);
		}
	}

	@Override
	public Channel[] getOpenedChannelsTo(InetAddress dest) {
		Channel[] openedChannels = getOpenedChannels();
		List<Channel> target = new ArrayList<Channel>();
		for (Channel channel : openedChannels) {
			if (channel.getRemoteAddress().getAddress().equals(dest)) {
				target.add(channel);
			}
		}
		return target.toArray(new Channel[0]);
	}

	@Override
	public void stop() {
		synchronized (stateLock) {
			if (state == STATE_STOPED) {
				return;
			}
			synchronized (openedChannels) {
				if (openedChannels.size() <= 0) {
					if (workerExecutorService != null) {
						workerExecutorService.shutdown();
					}
				}
			}
		}
	}

	public class ClientAdatper implements TransportAdapter {

		@Override
		public void channelClosed(ChannelImpl channel) {
			synchronized (stateLock) {
				synchronized (openedChannels) {
					openedChannels.remove(channel);
					if (openedChannels.size() <= 0) {
						if (state == STATE_STOPED) {
							if (workerExecutorService != null) {
								workerExecutorService.shutdown();
							}
						}
					}
				}
			}
		}

	}

}
